package com.mjf.loginfail_detect

import java.net.URL
import java.util

import com.mjf.dim.{LoginEvent, Warning}
import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.{CEP, PatternStream}
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 使用 CEP 实现 监测连续两次登录失败的用户
 *
 *
 * CEP:一个或多个由简单事件构成的事件流通过一定的规则匹配，然后输出用户想得到的数据 —— 满足规则的复杂事件
 *
 *  -目标：从有序的简单事件流中发现一些高阶特征
 *  -输入：一个或多个由简单事件构成的事件流
 *  -处理：识别简单事件之间的内在联系，多个符合一定规则的简单事件构成复杂事件
 *  -输出：满足规则的复杂事件
 *
 *
 * Pattern:处理事件的规则，被叫做“模式”
 *  -用于对输入流数据进行复杂事件规则定义，用来提取符合规则的事件序列
 *
 *  -个体模式：组成复杂规则的每一个单独的模式定义，就是“个体模式”
 *    +单例（singleton）模式：单例模式只接收一个事件
 *    +循环（looping）模式：接收多个事件
 *  -量词：可以在一个个体模式后追加量词，也就是指定循环次数
 *    +start.times(4) : 匹配出现4次
 *    +start.times(4).optional : 匹配出现0或4次
 *    +start.times(2,4) : 匹配出现2,3或4次
 *    +start.times(2,4).greedy : 匹配出现2,3或4次,并且尽可能多地重复匹配
 *    +start.oneOrMore : 匹配出现1次或者多次
 *    +start.timeOrMore(2).optional.greedy : 匹配出现0次、2次或者多次,并且尽可能多地重复匹配
 *  -条件：
 *    +每个模式都需要指定触发条件，作为模式是否接受事件进入的判断依据
 *    +CEP 中的个体模式主要通过调用 .where()  .or() 和 .until() 来指定条件
 *    +按不同的调用方式，可以分成以下几类：
 *      ~简单条件：通过 .where() 方法对事件中的字段进行判断筛选，决定是否接受该事件
 *      ~组合条件：将简单条件进行合并；.or() 方法表示或逻辑相连，where 的直接组合就是 and
 *      ~终止条件：如果使用了 oneOrMore 或者 oneOrMore.optional，建议使用 .until() 作为终止条件，以便清理状态
 *      ~迭代条件：
 *        #能够对模式之前所有接收的事件进行处理
 *        #调用 .where( (value, ctx) => {...} )，可以调用 ctx.getEventsForPattern(“name”)
 *
 *  -组合模式（Combining Patterns，也叫模式序列）：
 *    +很多个体模式组合起来，就形成了整个的模式序列
 *    +模式序列必须以一个“初始模式”开始：Pattern.begin[]()
 *  -不同的“近邻”模式
 *    +严格近邻：
 *      ~所有事件按照严格的顺序出现，中间没有任何不匹配的事件，由 .next() 指定
 *      ~例如：对于模式”a  next b”，事件序列 [a, c, b1, b2] 没有匹配
 *    +宽松近邻：
 *      ~允许中间出现不匹配的事件，由 .followedBy() 指定
 *      ~例如：对于模式”a followedBy b”，事件序列 [a, c, b1, b2] 匹配为 {a, b1}
 *    +非确定性宽松近邻：
 *      ~进一步放宽条件，之前已经匹配过的事件也可以再次使用，由 .followedByAny() 指定
 *      ~例如：对于模式”a followedByAny b”，事件序列 [a, c, b1, b2] 匹配为 {a, b1}，{a, b2}
 *    +不希望出现某种近邻关系：
 *      ~.notNext() : 不想让某个事件严格紧邻前一个事件发生
 *      ~.notFollowedBy() : 不想让某个事件在两个事件之间发生
 *  -注意：
 *    +所有模式序列必须以 .begin() 开始
 *    +模式序列不能以 .notFollowedBy() 结束
 *    +“not” 类型的模式不能被 optional 所修饰
 *    +此外，还可以为模式指定时间约束(within)，用来要求在多长时间内匹配有效
 *
 *  -模式组：将一个模式序列作为条件嵌套在个体模式里，成为一组模式
 *
 *
 *  -模式的检测：
 *    +指定要查找的模式序列后，就可以将其应用于输入流以检测潜在匹配
 *    +调用 CEP.pattern()，给定`输入流`和`模式`，就能得到一个 PatternStream
 *
 *  -匹配事件的提取：
 *    +创建 PatternStream 之后，就可以应用 select 或者 flatSelect 方法，从检测到的事件序列中提取事件了
 *    +select() 方法需要输入一个 PatternSelectFunction 作为参数，每个成功匹配的事件序列都会调用它
 *    +select() 以一个 Map[String, Iterable [IN]] 来接收匹配到的事件序列，其中 key 就是每个模式的名称，而 value 就是所有接收到的事件的 Iterable 类型
 *
 *  -超时事件的提取：当一个模式通过 within 关键字定义了检测窗口时间时，部分事件序列可能因为超过窗口长度而被丢弃
 *    +为了能够处理这些超时的部分匹配，select 和 flatSelect API 调用允许指定超时处理程序
 *    +超时处理程序会接收到目前为止由模式匹配到的所有事件，由一个 OutputTag 定义接收到的超时事件序列（详见 OrderPayDetect）
 *
 */
object LoginFailWithCEP {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据并处理
    val resource: URL = getClass.getResource("/LoginLog.csv")
    val loginEventStream: DataStream[LoginEvent] = env.readTextFile(resource.getPath)
      .map {
        line =>
          val dataArr: Array[String] = line.split(",")
          LoginEvent(dataArr(0).toLong, dataArr(1), dataArr(2), dataArr(3).toLong)
      }.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(3)) {
      override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
    })

    // 定义匹配的模式
    val loginFailPattern: Pattern[LoginEvent, LoginEvent] = Pattern
      .begin[LoginEvent]("firstFail").where(_.eventType == "fail") // 第一次登录失败
      .next("secondFail").where(_.eventType == "fail") // 第二次登录失败
      .within(Time.seconds(2))  // 在2秒钟之内检测匹配

    // 在分组之后的数据流上应用模式，得到一个PatternStream
    val patternStream: PatternStream[LoginEvent] = CEP.pattern(loginEventStream.keyBy(_.userId), loginFailPattern)

    // 将检测到的事件序列，转换输出报警信息
    val loginFailStream: DataStream[Warning] = patternStream.select(new LoginFailCEP())

    // 打印输出
    loginFailStream.print()

    env.execute("LoginFailWithCEP")

  }
}

// 将检测到的连续登录失败事件，包装成报警信息输出
class LoginFailCEP() extends PatternSelectFunction[LoginEvent, Warning] {
  override def select(map: util.Map[String, util.List[LoginEvent]]): Warning = {
    // map里存放的是匹配到的一组事件，key是定义好的事件模式名称
    val firstLoginFail: LoginEvent = map.get("firstFail").get(0)
    val secondLoginFail: LoginEvent = map.get("secondFail").get(0)
    Warning(firstLoginFail.userId, firstLoginFail.eventTime, secondLoginFail.eventTime, "login fail!")
  }
}